package com.gzx.plugin.tjzy.modular.blacklist.service.impl;

import cn.hutool.core.collection.CollUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.google.common.collect.Lists;
import com.gzx.plugin.tjzy.modular.blacklist.bo.GzxBlackListAddBo;
import com.gzx.plugin.tjzy.modular.blacklist.bo.GzxBlackListQueryBo;
import com.gzx.plugin.tjzy.modular.blacklist.dto.GzxBlackRespDto;
import com.gzx.plugin.tjzy.modular.blacklist.entity.GzxBlackList;
import com.gzx.plugin.tjzy.modular.blacklist.mapper.GzxBlackListMapper;
import com.gzx.plugin.tjzy.modular.blacklist.service.GzxBlackListDbService;
import com.gzx.plugin.tjzy.modular.blacklist.service.GzxBlackListService;
import com.gzx.plugin.tjzy.modular.blacklist.vo.GzxBlackListVo;
import com.gzx.plugin.tjzy.modular.call.ht.entity.GzxHtCallRecord;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;
import org.springframework.web.multipart.MultipartFile;
import vip.xiaonuo.auth.core.util.StpLoginUserUtil;
import vip.xiaonuo.common.consts.GzxTenantConstant;
import vip.xiaonuo.common.consts.IntConstant;
import vip.xiaonuo.common.mvc.service.impl.BaseServiceImpl;
import vip.xiaonuo.common.tenant.TenantContext;
import vip.xiaonuo.common.util.*;

import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

import static org.apache.commons.lang.StringEscapeUtils.escapeSql;

/**
 * @author xbl
 */
@Slf4j
@Service
@DS("black")
public class GzxBlackListServiceImpl extends BaseServiceImpl<GzxBlackListAddBo, GzxBlackListVo, GzxBlackListQueryBo, GzxBlackListMapper, GzxBlackList> implements GzxBlackListService {

    @Value("${black.xf.url}")
    private String url;

    @Value("${black.xf.ak}")
    private String ak;

    @Value("${black.xf.call-id}")
    private String callId;

    @Value("${black.xf.poly}")
    private String poly;

    //空号检测
//    @Value("${invalid.appId}")
    private String appId;

//    @Value("${invalid.appKey}")
    private String appKey;

//    @Value("${invalid.url}")
    private String invalidUrl;

    private final GzxBlackListDbService blackListDbService;

    public GzxBlackListServiceImpl(GzxBlackListMapper baseMapper, GzxBlackListDbService blackListDbService) {
        super(baseMapper);
        this.blackListDbService = blackListDbService;
    }

    @Override
    protected LambdaQueryWrapper<GzxBlackList> buildQueryWrapper(GzxBlackListQueryBo queryBo) {
        int searchType = queryBo.getSearchType();
        LambdaQueryWrapper<GzxBlackList> wrapper = Wrappers.<GzxBlackList>lambdaQuery()
                // 手机号
                .eq(StringUtils.isNotBlank(queryBo.getPhone()), GzxBlackList::getPhone, CommonCryptogramUtil.doSm1AesEncrypt(queryBo.getPhone()))
                // 按照创建日期范围查询
                .between(StringUtils.isNotBlank(queryBo.getBeginTime()) && StringUtils.isNotBlank(queryBo.getEndTime()), GzxBlackList::getCreateTime, queryBo.getBeginTime(), queryBo.getEndTime());
        // 根据 searchType 设置 type 查询条件，不传type的时候使用searchType限制查询调解，传type的时候使用type限制查询
        if (searchType == IntConstant.VALUE_1 && queryBo.getType() == IntConstant.VALUE_0) {
            // searchType=1 只能查询 type=1 或 type=2 的记录
            wrapper.in(GzxBlackList::getType, Arrays.asList(IntConstant.VALUE_1, IntConstant.VALUE_2));
        } else if (searchType == IntConstant.VALUE_2 && queryBo.getType() == IntConstant.VALUE_0) {
            // searchType=2 只能查询 type=3 的记录
            wrapper.eq(GzxBlackList::getType, IntConstant.VALUE_3);
        } else if (queryBo.getType() != IntConstant.VALUE_0) {
            wrapper.eq(GzxBlackList::getType, queryBo.getType());
        }

        return wrapper.orderByDesc(GzxBlackList::getCreateTime);
    }

    /**
     * 插入一个新的业务对象。
     *
     * @param gzxBlackListAddBo 业务对象
     */
    @Override
    public void insert(GzxBlackListAddBo gzxBlackListAddBo) {
        GzxBlackList blackList = MapstructUtils.convert(gzxBlackListAddBo, GzxBlackList.class);
        if (ObjectUtils.isEmpty(blackList)) {
            log.error("新增黑名单失败，参数为空 {}", gzxBlackListAddBo);
            return;
        }
        // 设置黑名单类型
        blackList.setType(gzxBlackListAddBo.getType());
        //手机号
        blackList.setPhone(CommonCryptogramUtil.doSm1AesEncrypt(blackList.getPhone()));
        blackList.setCreateUser(StpLoginUserUtil.getLoginUser().getId());
        blackList.setTenantId(StpLoginUserUtil.getLoginUser().getTenantId());
        baseMapper.insertIgnore(blackList);
    }

    /**
     * 批量校验手机号是不是黑名单
     */
    @DS("black")
    @Override
    public List<GzxBlackListVo> checkBlackList(List<String> phoneList) {
        List<GzxBlackListVo> rspList = new ArrayList<>();
        //入库的对象
        List<GzxBlackList> blackList = new ArrayList<>();

        //校验phoneList是不是空的
        if (CollUtil.isEmpty(phoneList)) {
            log.info("批量校验手机号是不是黑名单 为空");
            return rspList;
        }

        // 分批查询数据库，每次一千条
        List<List<String>> batchList = Lists.partition(new ArrayList<>(phoneList), 1000);

        for (List<String> batch : batchList) {
            List<GzxBlackListVo> batchResult = baseMapper.selectVoList(
                    Wrappers.<GzxBlackList>lambdaQuery()
                            .in(GzxBlackList::getPhone, batch)
                            .eq(GzxBlackList::getBelongingMonth, CommonDateUtil.getTodayStr())
            );
            if (CollUtil.isNotEmpty(batchResult)) {
                rspList.addAll(batchResult);
                batch.removeAll(batchResult.stream().map(GzxBlackListVo::getPhone).toList());
            }
            if (CollUtil.isNotEmpty(batch)) {
                Map<String, Object> headerMap = new HashMap<>();
                headerMap.put("appId", appId);
                headerMap.put("appKey", appKey);
                headerMap.put("mobile", String.join(",", batch));

                String resp = null;
                try {
                    resp = CommonHttpUtil.postByForm(invalidUrl, headerMap, "");
                    //json解析，入库
                    if (!ObjectUtils.isEmpty(resp)) {
                        //解析json
                        JSONArray arr = JSONObject.parseObject(resp, JSONArray.class);
                        for (int i = 0, size = arr.size(); i < size; i++) {
                            GzxBlackList black = new GzxBlackList();
                            JSONObject json = arr.getJSONObject(i);
                            //status为0的是空号
                            if ("0".equals(json.getString("status"))) {
                                black.setPhone(json.getString("mobile"));
                                //空号
                                black.setType(3);
                                black.setBelongingMonth(CommonDateUtil.getTodayStr());
                                blackList.add(black);
                            }
                        }
                        //入库
                        baseMapper.insertBatch(blackList);
                    } else {
                        log.error("停空号码为空：");
                    }
                } catch (Exception e) {
                    log.error("空号检测异常 {} ", e.getMessage());
                }
            }
        }

        // type = 0 的值过滤掉，0是白名单
        rspList = rspList.stream().filter(item -> !"0".equals(item.getType())).toList();

        return rspList;
    }

    //查询所有号码库黑名单（1：线路|业务，2：停机空号）
    @DS("black")
    @Override
    public List<GzxBlackListVo> queryAll(String type) {
        return baseMapper.selectVoList(Wrappers.<GzxBlackList>lambdaQuery().eq(GzxBlackList::getType, type));
    }

    //查询这个手机是不是在黑名单里
    @DS("black")
    @Override
    public boolean isBlackList(String phone) {
        //判断这个手机是不是在黑名单里
        GzxBlackListVo one = baseMapper.selectVoOne(Wrappers.<GzxBlackList>lambdaQuery().eq(GzxBlackList::getPhone, CommonCryptogramUtil.doSm1AesEncrypt(phone)).last("limit 1"));
        return !ObjectUtils.isEmpty(one);
    }

    /**
     * 校验黑名单
     *
     * @param phoneList
     */
    @DS("black")
    @Override
    public List<GzxBlackRespDto.DataResp> verifyBlack(List<String> phoneList) {
        if (CollUtil.isNotEmpty(phoneList)) {
            //查询手机号是否已经查过
            List<GzxBlackRespDto.DataResp> dataRespList = setRespList(phoneList);
            if (CollUtil.isNotEmpty(dataRespList)) {
                List<String> fromDbPhoneList = new ArrayList<>(
                        dataRespList.stream()
                                .map(GzxBlackRespDto.DataResp::getCallee)
                                .toList());
                phoneList.removeAll(fromDbPhoneList);
            }
            if (CollUtil.isNotEmpty(phoneList)) {
                Map<String, Object> bodyMap = new HashMap<>();
                bodyMap.put("ak", ak);
                bodyMap.put("callId", callId);
                bodyMap.put("callees", phoneList);
                bodyMap.put("poly", poly);
                bodyMap.put("sign", CommonCryptogramUtil.doMd5(ak + callId));
                try {
                    String resp = CommonHttpUtil.post(url, null, JSON.toJSONString(bodyMap), "南京先丰黑名单校验");
                    GzxBlackRespDto blackRespDto = JSONObject.parseObject(resp, GzxBlackRespDto.class);
                    if (blackRespDto != null && blackRespDto.getErrno() == 0 && CollUtil.isNotEmpty(blackRespDto.getData())) {
                        //异步入库
                        blackListDbService.saveBatch(blackRespDto.getData());
                        dataRespList.addAll(blackRespDto.getData());
                    }
                } catch (Exception e) {
                    log.error("南京先丰黑名单校验出现异常：", e);
                }
            }
            return dataRespList;
        }
        return List.of();
    }

    @DS("black")
    @Override
    public String importExcel(GzxBlackListAddBo bo, MultipartFile file) {
        long start = System.currentTimeMillis();
        //excel的数据
        List<GzxBlackListVo> excelVoList = CommonExcelUtil.importExcel(file, GzxBlackListVo.class);
        List<GzxBlackList> blackList = MapstructUtils.convert(excelVoList, GzxBlackList.class);

        if (ObjectUtils.isEmpty(blackList)) {
            return "Excel为空";
        }

        //去掉 blackList 中 phone 为空的数据
        blackList = blackList.stream().filter(v -> !ObjectUtils.isEmpty(v.getPhone())).toList();

        if (ObjectUtils.isEmpty(blackList)) {
            return "Excel为空";
        }

        //获取登录人信息
        String userId = StpLoginUserUtil.getLoginUser().getId();
        String tenantId = StpLoginUserUtil.getLoginUser().getTenantId();

        //批量处理每次1000条
        List<List<GzxBlackList>> bacthList = Lists.partition(blackList, IntConstant.VALUE_1000);
        for (List<GzxBlackList> blackBatch : bacthList) {
            String batchSql = blackBatch.stream().map(s -> String.format("('%s', '%s','%s', '%s', '%s')", escapeSql(CommonCryptogramUtil.doSm1AesEncrypt(s.getPhone())), escapeSql(String.valueOf(bo.getType())), null, escapeSql(tenantId), escapeSql(userId))).collect(Collectors.joining(", ")).replace("'null'", "null");
            baseMapper.insertBatchSql(batchSql);
            CommonSleepUtil.sleep(IntConstant.VALUE_100);
        }
        log.info("导入成功，共{}条，耗时{}秒", blackList.size(), System.currentTimeMillis() - start);
        return "成功";
    }

    //消费Kafka停机空号数据
    @Override
    public void asyncInsertRecord(List<ConsumerRecord<String, String>> recordList) {
        try {
            //禁用租户
            TenantContext.disableTenantFilter();
            //手机号不需要加密了，发到Kafka之前已经加密过了
            List<GzxHtCallRecord> htCallRecordList = getHtCallRecordList(recordList);
            List<GzxBlackList> blackBatch = new ArrayList<>();
            String userId = GzxTenantConstant.GZX_USER_ID;
            String tenantId = GzxTenantConstant.GZX_TENANT_ID;
            //把htCallRecordList的数据放到 blackBatch 中
            for (GzxHtCallRecord htCallRecord : htCallRecordList) {
                GzxBlackList blackList = new GzxBlackList();
                blackList.setPhone(htCallRecord.getCallee());
                //停机空号
                blackList.setType(IntConstant.VALUE_3);
                blackList.setCreateUser(userId);
                blackList.setTenantId(tenantId);
                blackBatch.add(blackList);
            }
            //批量处理每次1000条
            List<List<GzxBlackList>> bacthList = Lists.partition(blackBatch, IntConstant.VALUE_1000);
            for (List<GzxBlackList> black : bacthList) {
                String batchSql = black.stream().map(s -> String.format("('%s', '%s', '%s', '%s', '%s')", escapeSql(s.getPhone()), escapeSql(String.valueOf(IntConstant.VALUE_3)), null, escapeSql(tenantId), escapeSql(userId))).collect(Collectors.joining(", ")).replace("'null'", "null");
                log.info("批量插入停机空号数据：{}", batchSql);
                baseMapper.insertBatchSql(batchSql);
                CommonSleepUtil.sleep(IntConstant.VALUE_100);
            }
        } finally {
            //启用租户
            TenantContext.enableTenantFilter();
            TenantContext.clear();
        }
    }

    //解析为kafka数据为GzxHtCallRecord
    private List<GzxHtCallRecord> getHtCallRecordList(List<ConsumerRecord<String, String>> recordList) {
        String currentDate = CommonDateUtil.getCurrentTimeStr();
        List<GzxHtCallRecord> htCallRecordList = new ArrayList<>();
        for (ConsumerRecord<String, String> record : recordList) {
            GzxHtCallRecord htCallRecord = JSON.parseObject(record.value(), GzxHtCallRecord.class);
            htCallRecord.setCreatedTime(currentDate);
            htCallRecordList.add(htCallRecord);
        }
        return htCallRecordList;
    }

    private List<GzxBlackRespDto.DataResp> setRespList(List<String> phoneList) {
        List<String> encryptPhoneList = phoneList.stream().map(CommonCryptogramUtil::doSm1AesEncrypt).toList();
        List<GzxBlackList> blackList = baseMapper.selectList(Wrappers.<GzxBlackList>lambdaQuery()
                .and(lqw ->
                        lqw.eq(GzxBlackList::getBelongingMonth, CommonDateUtil.getTodayStr()).or().isNull(GzxBlackList::getBelongingMonth))
                .in(GzxBlackList::getType, 0, 1)
                .in(GzxBlackList::getPhone, encryptPhoneList));
        // 优先保留type为1的记录
        List<GzxBlackList> newList = blackList.stream()
                .collect(Collectors.toMap(
                        GzxBlackList::getPhone,  // 以手机号为key去重
                        Function.identity(),
                        (oldVal, newVal) -> oldVal.getType() == 1 ? oldVal : newVal  // 保留type=1的记录
                ))
                .values()
                .stream()
                .sorted(Comparator.comparingInt(GzxBlackList::getType).reversed())
                .toList();
        List<GzxBlackRespDto.DataResp> dataRespList = new ArrayList<>();
        for (GzxBlackList black : newList) {
            GzxBlackRespDto.DataResp dataResp = new GzxBlackRespDto.DataResp();
            dataResp.setCallee(black.getPhone());
            dataResp.setForbid(black.getType());
            dataRespList.add(dataResp);
        }
        return dataRespList;
    }

}
